package tasks.io.core.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tasks.io.core.ChannelContext;
import tasks.io.core.Packet;
import tasks.io.core.SystemTimer;
import tasks.io.core.TioConfig;
import tasks.io.core.lock.MapWithLock;
import tasks.io.core.thread.AbstractQueueTask;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;


public class HandlerTask extends AbstractQueueTask<Packet> {
    private static final Logger log = LoggerFactory.getLogger(HandlerTask.class);

    private ChannelContext channelContext = null;
    private TioConfig tioConfig = null;

    private final AtomicLong synFailCount = new AtomicLong();

    public HandlerTask(ChannelContext channelContext, Executor executor) {
        super(executor);
        this.channelContext = channelContext;
        tioConfig = channelContext.tioConfig;
        getQueue();
    }


    @Override
    public void runTask() {
        Packet packet = null;
        while ((packet = queue.poll()) != null) {
            handler(packet);
        }
    }

    /**
     * 处理packet
     */
    public void handler(Packet packet) {


        long start = SystemTimer.currTime;
        try {
            Integer synSeq = packet.getSynSeq();
            if (synSeq != null && synSeq > 0) {
                MapWithLock<Integer, Packet> syns = tioConfig.getWaitingResps();
                Packet initPacket = syns.remove(synSeq);
                if (initPacket != null) {
                    synchronized (initPacket) {
                        syns.put(synSeq, packet);
                        initPacket.notify();
                    }
                } else {
                    log.error("[{}]同步消息失败, synSeq is {}, 但是同步集合中没有对应key值", synFailCount.incrementAndGet(), synSeq);
                }
            } else {
                tioConfig.getTioHandler().handler(packet, channelContext);
            }
        } catch (Throwable e) {
            log.error(packet.logstr(), e);
        } finally {
            long end = SystemTimer.currTime;
            long iv = end - start;
            if (tioConfig.statOn) {
                channelContext.channelStat.handledPackets.incrementAndGet();
                channelContext.channelStat.handledBytes.addAndGet(packet.getByteCount());
                channelContext.channelStat.handledPacketCosts.addAndGet(iv);

                tioConfig.groupStat.handledPackets.incrementAndGet();
                tioConfig.groupStat.handledBytes.addAndGet(packet.getByteCount());
                tioConfig.groupStat.handledPacketCosts.addAndGet(iv);
            }


            if (tioConfig.getTioListener() != null) {
                try {
                    tioConfig.getTioListener().onAfterHandled(channelContext, packet, iv);
                } catch (Exception e) {
                    log.error(e.toString(), e);
                }
            }

        }
    }


    /**
     * The msg queue.
     */
    private final ConcurrentLinkedQueue<Packet> queue = new ConcurrentLinkedQueue<Packet>();

    @Override
    public ConcurrentLinkedQueue<Packet> getQueue() {

        return queue;
    }

}
